package org.example.kafkastudy.config;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;

import java.nio.ByteBuffer;
import java.util.Map;

public class SendTimeInterceptor<K, V> implements ProducerInterceptor<K, V> {
    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        System.out.println("SendTimeInterceptor...");
        if (record == null) { // 可能前面的拦截器已经拦截过了，会返回 null
            return null;
        }

        Headers headers = record.headers();
        // 将当前时间戳转换为字节数组
        byte[] timestampHeader = ByteBuffer.allocate(Long.BYTES).putLong(System.currentTimeMillis()).array();
        // 添加时间戳到 Header 中
        headers.add(new RecordHeader("sendTime", timestampHeader));

        return new ProducerRecord<>(record.topic(), record.partition(), record.key(), record.value(), headers);
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
